package com.cctc.rds.scheduler.client;

import com.cctc.rds.nrpc.api.client.config.RpcClientConfig;
import com.cctc.rds.scheduler.client.raft.NettyRpcClientHolder;
import com.cctc.rds.scheduler.client.raft.rpc.TaskRpcRequestImpl;
import com.cctc.rds.scheduler.client.schedule.impl.DefaultTaskScheduleClient;
import com.cctc.rds.scheduler.client.worker.TaskWokerScheduler;
import com.cctc.rds.scheduler.core.raft.TaskRpc;
import com.cctc.rds.scheduler.client.schedule.TaskScheduleClient;
import sun.misc.Signal;

import java.util.concurrent.atomic.AtomicReference;

/**
 * 任务延迟调度器客户端 [启动类]。
 */
public class SchedulerClientApplication {
    private static final AtomicReference<NettyRpcClientHolder> RPC_CLIENTS = new AtomicReference<>();
    private static final AtomicReference<TaskScheduleClient> TASK_SCHEDULER_CLIENT = new AtomicReference<>();
    private static final AtomicReference<TaskWokerScheduler> TASK_WORKER = new AtomicReference<>();

    static {
        Signal.handle(new Signal("INT"), signal -> {
            try {
                stop();
            } catch (Exception ignored) {
            }
        });
    }

    public synchronized static void start(String app, String hosts, long connectionTimeoutMs, long waitResponseTimeoutMs) {
        if (TASK_SCHEDULER_CLIENT.get() == null) {
            String[] ipPorts = hosts.split(",");
            RpcClientConfig[] configs = new RpcClientConfig[ipPorts.length];
            for (int i = 0; i < configs.length; i++) {
                String[] ipPort = ipPorts[i].split(":");
                configs[i] = RpcClientConfig.builder()
                        .remoteIp(ipPort[0])
                        .remotePort(Integer.valueOf(ipPort[1]))
                        .connectionTimeoutMs(connectionTimeoutMs)
                        .waitResponseTimeoutMs(waitResponseTimeoutMs)
                        .build();
            }
            NettyRpcClientHolder clientHolder = new NettyRpcClientHolder(configs);
            RPC_CLIENTS.set(clientHolder);
            clientHolder.start();

            TaskRpc taskRpc = new TaskRpcRequestImpl(clientHolder);
            TASK_SCHEDULER_CLIENT.set(new DefaultTaskScheduleClient(app, taskRpc));
            TASK_WORKER.set(new TaskWokerScheduler(app, taskRpc));
        }
    }

    public static TaskScheduleClient getTaskScheduleClient() {
        return TASK_SCHEDULER_CLIENT.get();
    }

    public static void stop() throws Exception {
        if (RPC_CLIENTS.get() != null) {
            RPC_CLIENTS.get().stop();
            RPC_CLIENTS.set(null);
        }
        TASK_SCHEDULER_CLIENT.set(null);
        if (TASK_WORKER.get() != null) {
            TASK_WORKER.get().close();
            TASK_WORKER.set(null);
        }
    }

}
